package com.atguigu.flink.chapter10;


import com.atguigu.flink.bean.LoginEvent;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @Author lzc
 * @Date 2022/9/8 9:02
 */
public class Flink05_High_Project_Login {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        KeyedStream<LoginEvent, Long> stream = env
            .readTextFile("input/LoginLog.csv")
            .map(new MapFunction<String, LoginEvent>() {
                @Override
                public LoginEvent map(String line) throws Exception {
                    String[] data = line.split(",");
                    return new LoginEvent(
                        Long.valueOf(data[0]),
                        data[1],
                        data[2],
                        Long.parseLong(data[3]) * 1000  // 把s变成ms
                    );
                }
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((event, ts) -> event.getEventTime())
            )
            .keyBy(LoginEvent::getUserId);
        // 1. 定义模式
        // 2s内连续两次登陆失败
        Pattern<LoginEvent, LoginEvent> pattern = Pattern
            .<LoginEvent>begin("fail")
            .where(new SimpleCondition<LoginEvent>() {
                @Override
                public boolean filter(LoginEvent value) throws Exception {
                    return "fail".equals(value.getEventType());
                }
            })
            .times(2)
            .consecutive()  // next
            .within(Time.milliseconds(2001));
        // 2. 把模式作用到流上, 得到一个模式流
        PatternStream<LoginEvent> ps = CEP.pattern(stream, pattern);
        // 3. 取出匹配到的数据
        /*ps
            .process(new PatternProcessFunction<LoginEvent, String>() {
                // 匹配成功一次, 执行一次
                @Override
                public void processMatch(Map<String, List<LoginEvent>> map,
                                         Context context,
                                         Collector<String> out) throws Exception {
                    String msg = map.get("fail").get(0).getUserId() + " 正在恶意登录";
                    out.collect(msg);
                }
            })
            .print();*/
    
        /*ps
            .select(new PatternSelectFunction<LoginEvent, String>() {
                @Override
                public String select(Map<String, List<LoginEvent>> map) throws Exception {
                    String msg = map.get("fail").get(0).getUserId() + " 正在恶意登录";
                    return msg;
                }
            })
            .print();
        */
        // 把恶意登录的记录找到, 放入到流中
        /*ps
            .select(new PatternSelectFunction<LoginEvent, LoginEvent>() {
                @Override
                public LoginEvent select(Map<String, List<LoginEvent>> map) throws Exception {
                    return null;
                }
            })
            .print();*/
    
        /*ps
            .flatSelect(new PatternFlatSelectFunction<LoginEvent, LoginEvent>() {
                @Override
                public void flatSelect(Map<String, List<LoginEvent>> map,
                                       Collector<LoginEvent> out) throws Exception {
                    List<LoginEvent> list = map.get("fail");
                
                    for (LoginEvent loginEvent : list) {
                        out.collect(loginEvent);
                    
                    }
                }
            })
            .print();*/
    
        ps
            .process(new PatternProcessFunction<LoginEvent, LoginEvent>() {
                @Override
                public void processMatch(Map<String, List<LoginEvent>> map,
                                         Context ctx,
                                         Collector<LoginEvent> out) throws Exception {
                    List<LoginEvent> list = map.get("fail");
                
                    for (LoginEvent loginEvent : list) {
                        out.collect(loginEvent);
                    
                    }
                }
            })
            .print();
        
        
        env.execute();
    }
}
